// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <tests/io/test_common.h>

#include <algorithm>
#include <cstdint>
#include <fstream>  // IWYU pragma: keep
#include <vector>

#ifndef _WIN32
#include <fcntl.h>
#endif

#include <nebula/core/buffer.h>
#include <nebula/io/file.h>
#include <nebula/io/memory.h>
#include <nebula/core/memory_pool.h>
#include <nebula/testing/ktest_util.h>
#include <nebula/future/future.h>

namespace nebula {

namespace io {

void AssertFileContents(const std::string& path, const std::string& contents) {
  ASSERT_OK_AND_ASSIGN(auto rf, ReadableFile::open(path));
  ASSERT_OK_AND_ASSIGN(int64_t size, rf->get_size());
  ASSERT_EQ(size, contents.size());

  ASSERT_OK_AND_ASSIGN(auto actual_data, rf->read(size));
  ASSERT_TRUE(actual_data->equals(Buffer(contents)));
}

bool FileExists(const std::string& path) { return std::ifstream(path.c_str()).good(); }

turbo::Status PurgeLocalFileFromOsCache(const std::string& path) {
#if defined(POSIX_FADV_WILLNEED)
  int fd = open(path.c_str(), O_WRONLY);
  if (fd < 0) {
    return turbo::io_error_with_errno_payload(errno, "open on ", path,
                            " to clear from cache did not succeed.");
  }
  int err = posix_fadvise(fd, 0, 0, POSIX_FADV_DONTNEED);
  if (err != 0) {
    return turbo::io_error_with_errno_payload(err, "fadvise on ", path,
                            " to clear from cache did not succeed");
  }
  err = close(fd);
  if (err == 0) {
    return turbo::OkStatus();
  }
  return turbo::io_error_with_errno_payload(err, "close on ", path, " to clear from cache did not succeed");
#else
  return turbo::unimplemented_error("posix_fadvise is not implemented on this machine");
#endif
}

turbo::Status ZeroMemoryMap(MemoryMappedFile* file) {
  constexpr int64_t kBufferSize = 512;
  static constexpr uint8_t kZeroBytes[kBufferSize] = {0};

  TURBO_RETURN_NOT_OK(file->Seek(0));
  int64_t position = 0;
  TURBO_MOVE_OR_RAISE(int64_t file_size, file->get_size());

  int64_t chunksize;
  while (position < file_size) {
    chunksize = std::min(kBufferSize, file_size - position);
    TURBO_RETURN_NOT_OK(file->write(kZeroBytes, chunksize));
    position += chunksize;
  }
  return turbo::OkStatus();
}

void MemoryMapFixture::TearDown() {
  for (auto path : tmp_files_) {
    TURBO_UNUSED(std::remove(path.c_str()));
  }
}

void MemoryMapFixture::CreateFile(const std::string& path, int64_t size) {
  ASSERT_OK(MemoryMappedFile::create(path, size));
  tmp_files_.push_back(path);
}

turbo::Result<std::shared_ptr<MemoryMappedFile>> MemoryMapFixture::InitMemoryMap(
    int64_t size, const std::string& path) {
  TURBO_MOVE_OR_RAISE(auto mmap, MemoryMappedFile::create(path, size));
  tmp_files_.push_back(path);
  return mmap;
}

void MemoryMapFixture::AppendFile(const std::string& path) { tmp_files_.push_back(path); }

class TrackedRandomAccessFileImpl : public TrackedRandomAccessFile {
 public:
  explicit TrackedRandomAccessFileImpl(io::RandomAccessFile* delegate)
      : delegate_(delegate) {}

  turbo::Status close() override { return delegate_->close(); }
  bool closed() const override { return delegate_->closed(); }
  turbo::Result<int64_t> tell() const override { return delegate_->tell(); }
  turbo::Status Seek(int64_t position) override { return delegate_->Seek(position); }
  turbo::Result<int64_t> read(int64_t nbytes, void* out) override {
    TURBO_MOVE_OR_RAISE(auto position, delegate_->tell());
    SaveReadRange(position, nbytes);
    return delegate_->read(nbytes, out);
  }
  turbo::Result<std::shared_ptr<Buffer>> read(int64_t nbytes) override {
    TURBO_MOVE_OR_RAISE(auto position, delegate_->tell());
    SaveReadRange(position, nbytes);
    return delegate_->read(nbytes);
  }
  bool supports_zero_copy() const override { return delegate_->supports_zero_copy(); }
  turbo::Result<int64_t> get_size() override { return delegate_->get_size(); }
  turbo::Result<int64_t> read_at(int64_t position, int64_t nbytes, void* out) override {
    SaveReadRange(position, nbytes);
    return delegate_->read_at(position, nbytes, out);
  }
  turbo::Result<std::shared_ptr<Buffer>> read_at(int64_t position, int64_t nbytes) override {
    SaveReadRange(position, nbytes);
    return delegate_->read_at(position, nbytes);
  }
  Future<std::shared_ptr<Buffer>> read_async(const io::IOContext& io_context,
                                            int64_t position, int64_t nbytes) override {
    SaveReadRange(position, nbytes);
    return delegate_->read_async(io_context, position, nbytes);
  }

  int64_t num_reads() const override { return read_ranges_.size(); }
  int64_t bytes_read() const override {
    int64_t sum = 0;
    for (const auto& range : read_ranges_) {
      sum += range.length;
    }
    return sum;
  }

  const std::vector<io::ReadRange>& get_read_ranges() const override {
    return read_ranges_;
  }

 private:
  io::RandomAccessFile* delegate_;
  std::vector<io::ReadRange> read_ranges_;

  void SaveReadRange(int64_t offset, int64_t length) {
    read_ranges_.emplace_back(io::ReadRange{offset, length});
  }
};

std::unique_ptr<TrackedRandomAccessFile> TrackedRandomAccessFile::create(
    io::RandomAccessFile* target) {
  return std::make_unique<TrackedRandomAccessFileImpl>(target);
}

}  // namespace io
}  // namespace nebula
